package cloud.xiguapi.ubas.analysis.advertising.model;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 广告数据黑名单自定义ProcessFunction实现类
 *
 * @author 大大大西西瓜皮🍉
 * date: 2021-5-16 下午 03:11
 * desc:
 */
public class BlacklistUserFilter
        extends KeyedProcessFunction<Tuple, ProvinceAdvertisingClickEvent, ProvinceAdvertisingClickEvent> {

    /**
     * 一天的毫秒数
     */
    public static final int DAY_MS = 24 * 60 * 60 * 1000;
    /**
     * 广告点击上限次数, 默认为10
     */
    private int countUpperBound = 10;
    /**
     * 定义状态, 保存当前用户对某一广告的点击次数
     */
    private ValueState<Long> countState;
    /**
     * 定义一个标志状态，保存到昂前用户是否已经被发送到黑名单的侧输出流中
     */
    private ValueState<Boolean> isSentState;

    public BlacklistUserFilter(int countUpperBound) {
        this.countUpperBound = countUpperBound;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        this.countState = getRuntimeContext()
                .getState(new ValueStateDescriptor<>("ad-count", Long.class, 0L));
        this.isSentState = getRuntimeContext()
                .getState(new ValueStateDescriptor<>("is-sent", Boolean.class, false));
    }

    @Override
    public void processElement(ProvinceAdvertisingClickEvent value, Context ctx, Collector<ProvinceAdvertisingClickEvent> out) throws Exception {
        // 判断当前用户对同一广告的点击次数, 达到上限直接过滤并在侧输出流报警
        Long currentCount = countState.value();

        // 判断是否是第一个数据, 若是则注册一个24h后的定时器
        if (currentCount == 0) {
            // 默认是以UTC时间, 若要以北京时间0点清空, 则需要减去8 * DAY_MS
            ctx.timerService().registerEventTimeTimer((ctx.timerService().currentProcessingTime() / DAY_MS + 1) * DAY_MS - 8 * DAY_MS);
        }

        // 判断是否报警
        if (currentCount >= countUpperBound) {
            // 达到上限, 未发送过报警
            if (!isSentState.value()) {
                // 更新状态
                isSentState.update(true);
                ctx.output(new OutputTag<BlacklistUserWarning>("blacklist") {
                           },
                        BlacklistUserWarning.builder()
                                .userId(value.getUserId())
                                .advertisingId(value.getAdvertisingId())
                                .warningMessage("click over " + countUpperBound + " times.")
                                .build());
            }
            // 不再执行下方操作
            return;
        }

        // 点击次数+1, 正常输出数据到主流中
        countState.update(currentCount + 1);
        out.collect(value);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<ProvinceAdvertisingClickEvent> out) throws Exception {
        // 定时器触发, 清空所有状态
        countState.clear();
        isSentState.clear();
    }
}
